package top.cyuw.simplerpc.remoting.codec;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import top.cyuw.simplerpc.compress.Compress;
import top.cyuw.simplerpc.compress.CompressSelector;
import top.cyuw.simplerpc.constant.RpcConstants;
import top.cyuw.simplerpc.extension.ExtensionLoader;
import top.cyuw.simplerpc.dto.RpcMessage;
import top.cyuw.simplerpc.serialize.Serializer;
import top.cyuw.simplerpc.serialize.SerializerSelector;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author chen
 * @date 2023/3/13 11:05
 */
public class RpcMessageEncoder extends MessageToByteEncoder<RpcMessage> {

    private static final AtomicInteger REQUEST_ID_COUNTER = new AtomicInteger(0);

    @Override
    protected void encode(ChannelHandlerContext ctx, RpcMessage rpcMessage, ByteBuf buf) {
        buf.writeBytes(RpcConstants.MAGIC_NUMBER);
        buf.writeByte(RpcConstants.VERSION);
        buf.writerIndex(buf.writerIndex() + RpcConstants.LENGTH_FULL_LENGTH);
        byte messageType = rpcMessage.getMessageType();
        buf.writeByte(messageType);
        buf.writeInt(REQUEST_ID_COUNTER.getAndIncrement());

        byte[] body = null;
        int fullLength = RpcConstants.LENGTH_HEAD;

        if (messageType == RpcConstants.MESSAGE_TYPE_RPC_REQUEST
                || messageType == RpcConstants.MESSAGE_TYPE_RPC_RESPONSE) {
            String serializerName = SerializerSelector.getSerializer();
            Serializer serializer = ExtensionLoader.of(Serializer.class).getExtension(serializerName);
            body = serializer.serialize(rpcMessage.getBody());

            String compressName = CompressSelector.getCompress();
            Compress compress = ExtensionLoader.of(Compress.class).getExtension(compressName);
            body = compress.compress(body);
        }

        if (body != null) {
            buf.writeBytes(body);
            fullLength += body.length;
        }

        int writerIndex = buf.writerIndex();
        buf.writerIndex(writerIndex - fullLength + RpcConstants.LENGTH_MAGIC_NUMBER + RpcConstants.LENGTH_VERSION);
        buf.writeInt(fullLength);
        buf.writerIndex(writerIndex);
    }

}
